package cn.anjubao.cp.api.config.mq;

import cn.anjubao.cp.api.pay.PaymentService;
import cn.anjubao.cp.api.pay.vo.PayNotifyVo;
import com.alibaba.fastjson.JSON;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeOrderlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerOrderly;
import org.apache.rocketmq.common.message.MessageExt;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.scheduling.concurrent.ThreadPoolTaskExecutor;
import org.springframework.stereotype.Component;

import java.util.List;

@Component
public class MessageHandler implements MessageListenerOrderly {

    private static final Logger LOG = LoggerFactory.getLogger(MessageHandler.class);

    @Autowired
    private ThreadPoolTaskExecutor executor;

    @Autowired
    private PaymentService paymentService;

    @Override
    public ConsumeOrderlyStatus consumeMessage(List<MessageExt> list, ConsumeOrderlyContext consumeOrderlyContext) {

        list.stream().forEach(item -> {
            executor.submit(() -> {
                try {
                    PayNotifyVo vo = JSON.parseObject(new String(item.getBody()), PayNotifyVo.class);
                    LOG.debug("收到rocketMq消息队列数据：{}", JSON.toJSONString(vo) );
                    paymentService.doPayNotify(vo);
                } catch (Exception e) {
                    e.printStackTrace();
                }
            });
        });
        return ConsumeOrderlyStatus.SUCCESS;
    }
}
